package org.orbis.communicate.rpc;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.rpc.Client;
import org.msgpack.rpc.error.RemoteError;
import org.orbis.communicate.FlagsResult;
import org.orbis.communicate.MainService;
import org.orbis.communicate.deleter.DeleterPackage;
import org.orbis.communicate.node.Node;
import org.orbis.search.SearchPackage;
import org.orbis.search.SearchResult;
import org.orbis.structure.ContentInfo;
import org.orbis.structure.LookupPointer;

/** This class defines communications with Messagepack-RPC.
 * 
 * @author iamyusk */
public class Communicator{
	private static final int BUFFERSIZE = 100;

	public ArrayType getInfo(String host, int port){
		Client client = null;
		ArrayType retInfo = null;

		try{
			client = new Client(host, port);
			PeerHandlerInterface interfce = client.proxy(PeerHandlerInterface.class);
			client.setRequestTimeout(3);

			// Informations
			retInfo = interfce.getInfo();
		}catch(UnknownHostException e){
			e.printStackTrace();
		}catch(ClassCastException e){
			System.out.println("Getting information from " + host + ":" + port + " didn't work well.");
			e.printStackTrace();
			return null;
		}

		return retInfo;
	}

	public ArrayList<Node> getNodeList(String host, int port) throws Exception{
		ArrayList<Node> nodeList = new ArrayList<Node>();
		Client client = null;
		Node newNode;
		try{
			client = new Client(host, port);
			PeerHandlerInterface interfce = client.proxy(PeerHandlerInterface.class);
			client.setRequestTimeout(5);

			ArrayType nodelist = interfce.nodeList();
			MessagePackObject[] objs = nodelist.asArray();
			for(int i = 0; i < objs.length; i += 2){
				newNode = Node.create(objs[i].asString(), objs[i + 1].asInt());
				nodeList.add(newNode);
			}
		}finally{
			if(client != null) client.close();
		}

		return nodeList;
	}

	public boolean register(Node host, Node myNode){
		// Test
//		try{
//			Client client = new Client(host.getAddr(), host.getPort());
//			System.out.printf("Client: %s:%s%n", host.getAddr(), host.getPort());
//			PeerHandlerInterface infce = client.proxy(PeerHandlerInterface.class);
//			client.setRequestTimeout(5);
//
//			System.out.printf("My node info.: %s:%s%n", myNode.getAddr(), myNode.getPort());
//			infce.register(myNode.getAddr(), myNode.getPort());
//		}catch(UnknownHostException e){
//			e.printStackTrace();
//		}catch(RemoteError e){
//			e.printStackTrace();
//		}

//		System.out.println("Worker: " + host.worker());
		boolean result = host.worker().register(myNode.getAddr(), myNode.getPort());
		return result;
//		return true;
	}

	public void indexContentRemote(Node node, ArrayType contents) throws Exception{
		node.setTimeout(5);
		node.worker().indexContent(contents);
	}

	public void indexTermRemote(Node node, ArrayType termInfo) throws Exception{
		node.setTimeout(5);
		node.worker().indexTerm(termInfo);
	}

	public FlagsResult[] searchRemote(Node node, SearchPackage pkg, int cluster, ContentInfo contentInfo)
		throws Exception{

		ArrayType arrays = node.worker().searchTerm(pkg.pack(), cluster);
		MessagePackObject[] objs = arrays.asArray();

		FlagsResult[] result = new FlagsResult[objs[0].asInt()];
		int subtotal = 1;
		for(int i = 0; i < result.length; i++){
			result[i].setID(objs[subtotal].asLong());
			result[i].setNewest(objs[subtotal + 1].asInt());
			subtotal += 2;

			boolean[] bools = new boolean[contentInfo.binaryFlagNum()];
			for(int j = 0; j < bools.length; j++){
				bools[i] = objs[subtotal + j].asBoolean();
			}
			result[i].setBinaryFlags(bools);
			subtotal += bools.length;

			int[] ints = new int[contentInfo.integerFlagNum()];
			for(int j = 0; j < ints.length; j++){
				ints[i] = objs[subtotal + j].asInt();
			}
			result[i].setIntFlags(ints);
			subtotal += ints.length;

			long[] longs = new long[contentInfo.longFlagNum()];
			for(int j = 0; j < longs.length; j++){
				longs[i] = objs[subtotal + j].asLong();
			}
			result[i].setLongFlags(longs);
			subtotal += longs.length;

			float[] floats = new float[contentInfo.floatFlagNum()];
			for(int j = 0; j < floats.length; j++){
				floats[i] = objs[subtotal + j].asFloat();
			}
			result[i].setFloatFlags(floats);
		}

		return result;
	}

	public boolean getDatas(int clusterID, Node node, Node myNode, MainService mains){
		node.setTimeout(50);
		int left = node.worker().getDataSize(clusterID, myNode.getAttributeHash(0L));

		if(left == -1) return false;
		for(int i = 0; i < left; i++){
			ArrayType arrayedValue =
				node.worker().getData(clusterID, myNode.getAttributeHash(0x8000000000000000L));
			if(i == 0) mains.setClusterPointInfo(arrayedValue);
			else{
				MessagePackObject[] objs = arrayedValue.asArray();
				mains.fromArrayType(objs[0].asInt(), objs[1].asInt(), objs[2].asInt(), objs[3].asInt(),
					objs[4].asByteArray());
			}
		}

		return true;
	}

	@Deprecated
	public HashMap<String, ArrayList<LookupPointer>> getTermList(int clusterID, Node node){
		HashMap<String, ArrayList<LookupPointer>> retMap = new HashMap<String, ArrayList<LookupPointer>>();
		ArrayList<LookupPointer> ptrList = null;
		MessagePackObject[] objs = null;

		node.setTimeout(50);
		int ID = node.worker().getDataSize(clusterID, 0);
		System.out.println("ID: " + (ID & 0xFF));

		ArrayType datas = node.worker().getData(ID, 0);
		objs = datas.asArray();
		if(!objs[0].isBooleanType()){
			ptrList = new ArrayList<LookupPointer>();

			for(int j = 0; 3 * j + 3 < objs.length; j++)
				ptrList.add(LookupPointer.valueOf(objs[3 * j + 2].asInt(), objs[3 * j + 3].asInt()));

			retMap.put(objs[0].asString(), ptrList);
		}
		if(retMap.size() % 100 == 0) System.out.printf("\rGot %8d data", retMap.size());

		System.out.printf("\rGot %8d data\nEnd of recieving data.\n", retMap.size());
		node.worker().endDataSend(ID);

		return retMap;
	}

	/** get term list.
	 * 
	 * @param host ... IP address, it works as a key
	 * @param ptr ... pointer which starts reading invert index
	 * @return */
	@Deprecated
	public LinkedHashMap<Long, Integer> getInvertIndexList(Node node, LookupPointer ptr){
		LinkedHashMap<Long, Integer> invertIndices = new LinkedHashMap<Long, Integer>();

		while(!ptr.equals(LookupPointer.EMPTY)){
			// Get invert index
			ArrayType list = node.worker().getList(ptr.block(), ptr.point(), 0L, BUFFERSIZE);
			MessagePackObject[] hashObjs = list.asArray();
			if(hashObjs.length >= 3){
				// FIXME: Error in index
				ptr.modify(hashObjs[0].asInt(), hashObjs[1].asInt());
				for(int k = 3; k + 1 < hashObjs.length; k += 2)
					invertIndices.put(Long.valueOf(hashObjs[k].asLong()),
						Integer.valueOf(hashObjs[k + 1].asInt()));
			}
		}

		return invertIndices;
	}

	public SearchResult getContentSet(long ID, int contentLength, int appendixLength, Node node, int cluster){
		// node.setTimeout(3);
		MessagePackObject obj = node.worker().getContent(ID, cluster);
		if(!obj.isArrayType()) return null;
		MessagePackObject[] array = obj.asArray();
		String[] content = new String[contentLength];
		String[] appendix = new String[appendixLength];

		int i;
		for(i = 0; i < content.length; i++)
			content[i] = array[i].asString();
		for(i = 0; i < appendix.length; i++)
			appendix[i] = array[i + content.length].asString();

		return SearchResult.makeResult(ID, 0.0, content, appendix);
	}

	public boolean isLive(Node node) throws Exception{
		boolean retb = false;
		node.setTimeout(3);
		if(node.worker().isLive() != null) retb = true;

		return retb;
	}

	public void finalizeProcess(Node remoteNode, Node ownNode){
		try{
			remoteNode.worker().deleteNode(ownNode.getAddr(), ownNode.getPort());
		}finally{
			if(remoteNode.client() != null) remoteNode.closeConnection();
		}
	}

	public void deleteContent(Node node, DeleterPackage pkg){
		node.worker().deleteContent(pkg.pack(), pkg.cluster());
	}

	public void synchronizeClock(Node node){
		node.worker().synchronizeClock();
	}

	public void stopIndexing(Node node){
		node.worker().stopIndexing();
	}

	public void resumeIndexing(Node node){
		node.worker().resumeIndexing();
	}

	public void backupLatest(Node node){
		node.worker().backupLatest();
	}

	public void revertLatest(Node node){
		node.worker().revertLatest();
	}

	public void suspend(Node node, Node ownNode, long suspendTime, String reason){
		node.worker().suspend(ownNode.getAddr(), ownNode.getPort(), suspendTime, reason);
	}

	public void resume(Node node, Node target){
		node.worker().resume(target.getAddr(), target.getPort());
	}
}
